package org.jxkj.util

import java.util
import java.util.Properties

import org.apache.hadoop.hbase.TableName
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import org.joda.time.DateTime
import org.joda.time.format.{DateTimeFormat, DateTimeFormatter}
import org.jxkj.bean.PointData
import org.jxkj.data.hbase.HBaseDaoWideImpl
import org.jxkj.utils.{HBaseUtils, ParseUtils}
import org.jxkj.webService.{ArrayOfString, WebServiceSoap}
import org.jxkj.webapi.WebAPI

import scala.collection.JavaConversions._
import scala.collection.mutable

object DataInsert {

  var connection: Connection = null
  var region: Int = 20
  var properties: Properties = null;

  /**
    * 初始化一张新表
    *
    */
  def initialTable() = {
    //val hbase= new IDaoProvider()
    properties = PropertiesUtils.initial("hbase.properties")
    region = properties.getProperty("region").toInt
    connection = HBaseDaoWideImpl.getConnection
    /* if (!HBaseUtils.isExistTable(connection, tableName)) {
       HBaseUtils.createTable(connection, tableName, region, "CF")
     }*/
  }


  /**
    * 通过webapi抓取数据  @未使用
    *
    * @param pointName
    * @param start yyyy/MM/dd HH:mm:ss
    * @param end   yyyy/MM/dd HH:mm:ss
    */
  @Deprecated
  def waCollect(pointName: String, pointId: Int, start: String, end: String, webapi: WebAPI): util.ArrayList[Put] = {
    var ori: util.List[PointData] = null;
    //时间跨度小于一天，直接调用webapi
    ori = webapi.getHistRaw(pointName, start, end)
    var completedMap = mutable.Map[Int, String]()
    val num: Int = ori.size()
    if (num > 0) {
      for (item <- ori) {
        completedMap += (item.getTimestamp.toInt -> item.getValue)
      }
      val puts: util.ArrayList[Put] = addList(completedMap, pointName, pointId)
      return puts
    }
    return null
  }


  /**
    * 通过webservice抓取数据
    *
    * @param pointName
    * @param start yyyy/MM/dd HH:mm:ss
    * @param end   yyyy/MM/dd HH:mm:ss
    */
  def wsCollect(pointName: String, pointId: Int, start: String, end: String, soap: WebServiceSoap) = {
    val year: String = start.substring(0, 4)
    val tableName = properties.getProperty("table_" + year)
    val value: ArrayOfString = soap.getHistValue(pointName, start, end)
    val ori: util.List[String] = value.getString
    var map = mutable.Map[Int, String]()
    if (ori.size() > 0) {
      for (item <- ori) {
        val fields: Array[String] = item.split(",")
        map += (ParseUtils.getTimestamp(fields(0)) -> fields(1))
      }
      val puts: util.ArrayList[Put] = addList(map, pointName, pointId)
      commit(puts, tableName)
    }
  }


  /**
    * 批量提交
    * @param cacheList
    */
  @throws[Exception]
  def commit(cacheList: util.ArrayList[Put], tableName: String): Unit = {
    val listener: BufferedMutator.ExceptionListener = new BufferedMutator.ExceptionListener() {
      override def onException(e: RetriesExhaustedWithDetailsException, mutator: BufferedMutator): Unit = {
        var i = 0
        while ( {
          i < e.getNumExceptions
        }) {
          {
            i += 1;
            i - 1
          }
        }
      }
    }
    val params: BufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf(tableName)).listener(listener)
    params.writeBufferSize(5 * 1024 * 1024)
    if (connection == null) connection = HBaseDaoWideImpl.getConnection
    val mutator: BufferedMutator = connection.getBufferedMutator(params)
    try {
      mutator.mutate(cacheList)
      mutator.flush()
    } finally mutator.close()
  }


  def getMutator(start: String): BufferedMutator = {
    val year: String = start.substring(0, 4)
    val tableName = properties.getProperty("table_" + year)
    val listener: BufferedMutator.ExceptionListener = new BufferedMutator.ExceptionListener() {
      override def onException(e: RetriesExhaustedWithDetailsException, mutator: BufferedMutator): Unit = {
        var i = 0
        while ( {
          i < e.getNumExceptions
        }) {
          {
            i += 1;
            i - 1
          }
        }
      }
    }
    val params: BufferedMutatorParams = new BufferedMutatorParams(TableName.valueOf(tableName)).listener(listener)
    //自动提交间隔
    //params.setWriteBufferPeriodicFlushTimeoutMs(3000)
    //params.setWriteBufferPeriodicFlushTimerTickMs(1000)
    params.writeBufferSize(5 * 1024 * 1024)
    if (connection == null) connection = HBaseDaoWideImpl.getConnection
    val mutator: BufferedMutator = connection.getBufferedMutator(params)
    return mutator
  }


  def addList(datas: mutable.Map[Int, String], pointName: String, pointId: Int): util.ArrayList[Put] = {
    val sdf: DateTimeFormatter = DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss")
    val toHour: DateTimeFormatter = DateTimeFormat.forPattern("yyyyMMddHH")
    val regionCode: String = HBaseUtils.genRegionCode(pointId, 20)

    val cacheList = new util.ArrayList[Put]()
    for (data <- datas) {
      val timestamp = data._1
      val time: String = ParseUtils.getDate(timestamp)
      val value: String = data._2
      val rowkeyTime: String = DateTime.parse(time, sdf).toString("yyyMMddHH") //生成rowkey
      val rowkey: String = HBaseUtils.genRowKey(regionCode, String.format("%06d", Int.box(pointId)), rowkeyTime)
      println("rowkey--"+rowkey)
      //插入rowkey
      val str: String = DateTime.parse(rowkeyTime, toHour).toString("yyyy/MM/dd HH:mm:ss")
      val timestampStart = ParseUtils.getTimestamp(str)

      val put = new Put(Bytes.toBytes(rowkey))
      put.addColumn(Bytes.toBytes("CF"), Bytes.toBytes((timestamp - timestampStart) + ""), Bytes.toBytes(value))
      //设置写WAL （Write-Ahead-Log）的级别
      put.setDurability(Durability.SKIP_WAL)
      cacheList.add(put)
    }

    return cacheList
  }

  def addList2(datas: mutable.Map[Integer, String], pointName: String, pointId: Int): util.ArrayList[Put] = {
    val sdf: DateTimeFormatter = DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss")
    val toHour: DateTimeFormatter = DateTimeFormat.forPattern("yyyyMMddHH")
    val regionCode: String = HBaseUtils.genRegionCode(pointId, 20)

    val cacheList = new util.ArrayList[Put]()
    for (data <- datas) {
      val timestamp = data._1
      val time: String = ParseUtils.getDate(timestamp.longValue())
      val value: String = data._2
      val rowkeyTime: String = DateTime.parse(time, sdf).toString("yyyMMddHH") //生成rowkey
      val rowkey: String = HBaseUtils.genRowKey(regionCode, String.format("%06d", Int.box(pointId)), rowkeyTime)
      println("rowkey--"+rowkey)
      //插入rowkey
      val str: String = DateTime.parse(rowkeyTime, toHour).toString("yyyy/MM/dd HH:mm:ss")
      val timestampStart = ParseUtils.getTimestamp(str)

      val put = new Put(Bytes.toBytes(rowkey))
      put.addColumn(Bytes.toBytes("CF"), Bytes.toBytes((timestamp - timestampStart) + ""), Bytes.toBytes(value))
      //设置写WAL （Write-Ahead-Log）的级别
      put.setDurability(Durability.SKIP_WAL)
      cacheList.add(put)
    }

    return cacheList
  }

  /**
    * 获得rowkey的方法
    * @param pointId
    * @param time
    * @return
    */
  private def getRowkey(pointId:Int,regionCode:String,time:String):String = {
    val rowkeyTime: String = DateTime.parse(time, DateTimeFormat.forPattern("yyyy/MM/dd HH:mm:ss")).toString("yyyMMddHH") //生成rowkey
    val rowkey: String = HBaseUtils.genRowKey(regionCode, String.format("%06d", Int.box(pointId)), rowkeyTime)
    rowkey
  }
}
